package com.cbh.amadeus.client.core;

import com.cbh.amadeus.client.handler.*;
import com.cbh.amadeus.client.receiver.component.SubscriberComponent;
import com.cbh.amadeus.client.utils.ChannelReference;
import com.cbh.amadeus.common.handler.ExceptionHandler;
import com.cbh.amadeus.common.handler.HeartBitHandler;
import com.cbh.amadeus.common.handler.IdentifyHandler;
import com.cbh.amadeus.common.protocol.AmadeusMessageProtocol;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * @author ChenBiHui
 * @date 2025/4/27
 */
@Slf4j
public class Subscriber implements InitializingBean, DisposableBean {

    public static final EventLoopGroup DEFAULT = new DefaultEventLoopGroup(
            Runtime.getRuntime().availableProcessors(),
            new DefaultThreadFactory("amadeus sd")
    );
    public static final NioEventLoopGroup WORK = new NioEventLoopGroup(
            Runtime.getRuntime().availableProcessors(),
            new DefaultThreadFactory("amadeus sw")
    );
    @Resource
    private SubscriberComponent component;

    public static void connect(SubscriberComponent component, Integer sort) {
        Bootstrap client = new Bootstrap();
        ChannelFuture future = client.group(WORK)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) {
                        ch.pipeline()
                                .addLast(AmadeusMessageProtocol.frameDecoder())
                                .addLast(AmadeusMessageProtocol.protocol())
                                .addLast(new IdleStateHandler(component.getConfig().getReaderIdleTimeSeconds(), component.getConfig().getWriterIdleTimeSeconds(), 0))
                                .addLast(new IdentifyHandler())
                                .addLast(new HeartBitHandler(component.getConfig().getReaderIdle(), component.getConfig().getWriterIdle()))
                                .addLast(new ReconnectHandler(component, sort))
                                .addLast(new RegisteredHandler())
                                .addLast(new ReceiveHandler(component))
                                .addLast(new ReleaseHandler(component))
                                .addLast(new RejectHandler(component))
                                .addLast(new OnlyConnectBackHandler())
                                .addLast(new ExceptionHandler())
                        ;
                    }
                })
                .connect(component.getConfig().getHost(), component.getConfig().getPort());
        future.addListener(item -> {
            if (item.isSuccess()) {
                log.info(
                        "===== Amadeus Message center client is ready connect to the [{}:{}] =====",
                        component.getConfig().getHost(),
                        component.getConfig().getPort()
                );
                String name = ChannelReference.compose(component.getConfig().getSubscriberName(), sort);
                component.getRegister().register(future.channel(), name, sort);
            } else {
                log.info(
                        "===== Amadeus Message center server is not ready will try to reconnect to the [{}:{}] =====",
                        component.getConfig().getHost(),
                        component.getConfig().getPort()
                );
                future.channel()
                        .eventLoop()
                        .schedule(() -> Subscriber.connect(component, sort), component.getConfig().getReconnectTimeSeconds(), TimeUnit.SECONDS);
            }
        });
    }

    @Override
    public void afterPropertiesSet() {
        ChannelReference.loading(component);
        for (int sort = 0; sort < ChannelReference.channelCount(); sort++) {
            Subscriber.connect(component, sort);
        }
    }

    @Override
    public void destroy() {
        DEFAULT.shutdownGracefully();
        WORK.shutdownGracefully();
    }
}
